package com.woody.aggregate.application.service.shop.es;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.woody.aggregate.application.service.shop.es.strategy.ShopInfoConverterStrategy;
import com.woody.aggregate.application.service.shop.es.strategy.ShopInfoConverterStrategyFactory;
import com.woody.aggregate.common.consts.RedisLockKeyConst;
import com.woody.aggregate.common.consts.es.AiShopIndexIdPrefix;
import com.woody.aggregate.common.consts.es.ShopChangeTypeEnum;
import com.woody.aggregate.common.enums.ShopPropertyTypeEnum;
import com.woody.aggregate.domain.saas.SaasShopConfigManager;
import com.woody.aggregate.infrastructure.saas.entity.SaasShopConfig;
import com.woody.aggregate.infrastructure.shop.es.esdoc.ShopInfo;
import com.woody.aggregate.infrastructure.shop.es.esmapper.ShopInfoEsMapper;
import com.woody.aggregate.shop.enums.es.EsShopTypeEnum;
import com.woody.base.exception.BaseException;
import com.woody.base.model.Response;
import com.woody.base.user.api.dto.query.ShopIdsDTO;
import com.woody.base.user.api.vo.BaseMerchantVO;
import com.woody.base.user.api.vo.ShopVO;
import com.woody.base.user.feign.merchant.MerchantFeign;
import com.woody.base.user.feign.merchant.ShopFeign;
import com.woody.base.util.ResponseUtil;
import com.woody.common.redis.lock.TryLock;
import com.woody.common.redis.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.dromara.easyes.annotation.IndexName;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.concurrent.TimeUnit;

/**
 * &copy; All rights Reserved
 *
 * @author wénkǎi.zhāng 2025-04-07
 */
@Service
@Slf4j
public class EsShopSyncService {

    @Autowired
    private ShopInfoEsMapper shopInfoEsMapper;

    @Autowired
    private SaasShopConfigManager saasShopConfigManager;

    @Autowired
    private ShopFeign shopFeign;

    @Autowired
    private MerchantFeign merchantFeign;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 同步门店信息到es
     *
     * @param shopIds 建议单次不超过2000，一是大量查询可能导致内存占用过高，二是方法内部有通过in条件从数据库查询商圈配置
     */
    public void syncShopInfo(List<Long> shopIds) {
        if (CollectionUtils.isEmpty(shopIds)) {
            return;
        }

        // 尝试获取分布式锁，允许等待10秒
        boolean lockAcquired = false;
        try {
            log.info("门店信息同步到ES，开始尝试获取分布式锁");
            // 尝试在10秒内获取锁，锁的最长持有时间为30分钟
            lockAcquired = redissonClient.getLock(RedisLockKeyConst.ES_SHOP_SYNC_LOCK).tryLock(10, 1800, TimeUnit.SECONDS);

            if (!lockAcquired) {
                // 如果10秒内未能获取到锁，记录日志并保存任务信息，避免消息丢失
                throw new RuntimeException("同步门店到ES获取分布式锁失败，异常重试");
            }

            log.info("开始同步门店信息到ES，请求门店ID数量: {}，门店ID列表: {}", shopIds.size(), shopIds);
            long startTime = System.currentTimeMillis();

            /*
             * 一、数据准备：
             * 1. 数据库最新门店信息
             * 2. ES中门店信息
             * 3. 主门店商户配置
             * 4. 商圈配置
             */

            // 1. 根据门店id批量查询门店信息信息
            List<ShopVO> dbShopInfos = batchQueryShopInfoByIds(shopIds);


            // 检查是否成功获取到门店数据
            if (CollectionUtils.isEmpty(dbShopInfos)) {
                log.info("查询门店信息为空");
                return;
            }

            log.info("最新门店信息:{}", JSON.toJSONString(dbShopInfos));

            List<Long> finalShopIds = dbShopInfos.stream().map(ShopVO::getShopId).collect(Collectors.toList());
            // finalShopIds与入参取差集，finalShopIds里缺少的打印警告日志
            List<Long> missingShopIds = shopIds.stream().filter(id -> !finalShopIds.contains(id)).collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(missingShopIds)) {
                log.warn("以下门店ID未查询到门店数据: {}", missingShopIds);
            }

            // 2. 根据门店id从ES中查询门店信息
            List<String> esIds = finalShopIds.stream().map(AiShopIndexIdPrefix::getShopIndexId).collect(Collectors.toList());
            List<ShopInfo> esShopInfos = shopInfoEsMapper.selectBatchIds(esIds);
            log.info("ES门店信息:{}", JSON.toJSONString(esShopInfos));

            // 创建两个Map，存放本次同步涉及到的主门店商户配置和商圈配置
            // 3. 主门店配置 <主门店id, BaseMerchantVO>
            Map<Long, BaseMerchantVO> merchantMap = new HashMap<>();

            // 4. 商圈配置 <主门店id, SaasShopConfigDTO>
            Map<Long, SaasShopConfig> saasShopConfigMap = new HashMap<>();

            // 收集精选门店的主门店id，用于查询商户配置和商圈配置
            Set<Long> mainShopIds = new HashSet<>();

            for (ShopVO dbShopInfo : dbShopInfos) {
                // 当前是主门店
                if (ShopPropertyTypeEnum.MAIN.getType().equals(dbShopInfo.getProperty())) {
                    mainShopIds.add(dbShopInfo.getShopId());
                } else if (ShopPropertyTypeEnum.CHILD.getType().equals(dbShopInfo.getProperty()) && dbShopInfo.getParentId() != null && dbShopInfo.getParentId() > 0) {
                    // 当前是子门店
                    mainShopIds.add(dbShopInfo.getParentId());
                }
            }
            log.info("相关主门店ID列表: {}", mainShopIds);

            // 如果mainShopIds不为空，说明本次同步涉及到了精选门店，否则说明db中全部都是普通门店（es中不一定是）
            if (CollectionUtils.isNotEmpty(mainShopIds)) {
                // 通过merchantFeign查询商户配置
                ShopIdsDTO shopIdsDTO = new ShopIdsDTO();
                shopIdsDTO.setShopIds(new ArrayList<>(mainShopIds));
                Response<List<BaseMerchantVO>> merchantListResp = merchantFeign.getMerchantListByShopIds(shopIdsDTO);
                List<BaseMerchantVO> merchantList = ResponseUtil.getSucRes(merchantListResp);
                if (CollectionUtils.isNotEmpty(merchantList)) {
                    merchantMap.putAll(merchantList.stream().collect(Collectors.toMap(BaseMerchantVO::getShopId, Function.identity(), (v1, v2) -> v1)));
                    log.info("主门店商户配置信息: {}", JSON.toJSONString(merchantMap));
                }
                if(mainShopIds.size() != merchantMap.size()) {
                    log.warn("获取到商户配置数量与查询的主门店数量不一致");
                }

                // 查询本地数据库，获取商圈配置
                List<SaasShopConfig> saasShopConfigs = saasShopConfigManager.getByShopIds(mainShopIds);
                if (CollectionUtils.isNotEmpty(saasShopConfigs)) {
                    saasShopConfigMap.putAll(saasShopConfigs.stream().collect(Collectors.toMap(SaasShopConfig::getShopId, Function.identity(), (v1, v2) -> v1)));
                    log.info("商圈配置信息：{}", JSON.toJSONString(saasShopConfigMap));
                }
            }

            // 二、操作分类
            /*
             * 门店信息变更时，主要关注property字段的变化，而property有下列12中情形，只需要关注：
             * 1. 第7种从1 -> 0的情况，即：主门店变成了普通门店，相应的商圈需要删除
             * 2. 第8种从1 -> 1的情况，更新子门店的授权结束时间
             *
             * 由于不会对比db里哪些字段发生了变化，门店信息更新时直接更新ES全部字段，因此门店更新通过新增覆盖的方式。
             *
             * 下表中，只有前3行是新增门店，下面都是更新，但也通过insert的方式覆盖原文档。
             * 只有第8行的情形，单独更新授权结束时间字段
             *
             * ｜ 情形 ｜ES中的门店类型  ｜ db中的门店类型  ｜ 同步方式
             * ｜-----｜--------------｜-------------- ｜--------------
             * ｜  1  ｜              ｜       0       ｜ 新增门店
             * ｜  2  ｜              ｜       1       ｜ 新增门店
             * ｜  3  ｜              ｜       2       ｜ 新增门店
             * ｜  4  ｜     0        ｜       0       ｜ 新增门店
             * ｜  5  ｜     0        ｜       1       ｜ 新增门店
             * ｜  6  ｜     0        ｜       2       ｜ 新增门店
             * ｜  7  ｜     1        ｜       0       ｜ 新增门店、删除商圈
             * ｜  8  ｜     1        ｜       1       ｜ 新增门店，如果授权结束时间变了，更新子门店的授权结束时间
             * ｜  9  ｜     1        ｜       2       ｜ 新增门店
             * ｜  10 ｜     2        ｜       0       ｜ 新增门店
             * ｜  11 ｜     2        ｜       1       ｜ 新增门店
             * ｜  12 ｜     2        ｜       2       ｜ 新增门店
             */


            // 初始化不同逻辑的操作类型的列表
            Map<ShopChangeTypeEnum, List<ShopVO>> shopChangeTypeMap = new HashMap<>();
            shopChangeTypeMap.put(ShopChangeTypeEnum.ADD, new ArrayList<>());

            // 要删除的商圈ES文档id列表
            List<String> deleteEsIds = new ArrayList<>();

            // 需要更新授权结束时间的主门店配置
            // <主门店id, 授权结束时间秒级时间戳>
            Map<Long, Integer> authEndTimeMap = new HashMap<>();

            // 将ES中的门店信息转换为Map，方便后续使用
            Map<Long, ShopInfo> esShopInfoMap = esShopInfos.stream()
                    .collect(Collectors.toMap(
                            shopInfo -> Long.parseLong(shopInfo.getShopId()),
                            Function.identity(),
                            (v1, v2) -> v1
                    ));

            // 遍历数据库中的门店信息，分类处理
            for (ShopVO dbShopInfo : dbShopInfos) {
                shopChangeTypeMap.get(ShopChangeTypeEnum.ADD).add(dbShopInfo);

                if (!esShopInfoMap.containsKey(dbShopInfo.getShopId())) {
                    // ES里没有，属于新增门店，无需执行后续逻辑
                    continue;
                }

                Long shopId = dbShopInfo.getShopId();
                // 数据库里的门店类型
                ShopPropertyTypeEnum dbPropertyType = ShopPropertyTypeEnum.fromType(dbShopInfo.getProperty()).orElse(ShopPropertyTypeEnum.NORMAL);

                // 当前门店在ES里是不是主门店，如果是主门店需要判断上述第7、8两种情形
                ShopInfo esShopInfo = esShopInfoMap.get(shopId);
                if (!EsShopTypeEnum.MAIN.getType().equals(esShopInfo.getShopType())) {
                    continue;
                }

                // 第7种情况判断，es里当前是主门店，db里是普通门店，需要删除商圈
                if (ShopPropertyTypeEnum.NORMAL.equals(dbPropertyType)) {
                    String bdIndexId = AiShopIndexIdPrefix.getBdIndexId(shopId);
                    deleteEsIds.add(bdIndexId);
                    log.info("主门店变为普通门店，将删除对应商圈，门店ID: {}, 商圈ID: {}", shopId, bdIndexId);
                }

                // 第8种情况判断，db里也是主门店，但如果商户配置里的授权结束时间变了，更新子门店的授权结束时间
                if (ShopPropertyTypeEnum.MAIN.equals(dbPropertyType)) {
                    BaseMerchantVO merchantVO = merchantMap.get(shopId);
                    int dbAuthEndTime = merchantVO != null ? (int) (merchantVO.getAuthEndDate().getTime() / 1000) : 0;
                    Integer esAuthEndTime = esShopInfo.getAuthEndTime();
                    if (!Objects.equals(esAuthEndTime, dbAuthEndTime)) {
                        // 记录主门店的授权结束时间，稍后通过主门店为条件，批量更新子门店的授权结束时间
                        authEndTimeMap.put(shopId, dbAuthEndTime);
                        log.info("主门店授权时间变更，将更新子门店授权结束时间，门店ID: {}, 原授权时间: {}, 新授权时间: {}",shopId, esAuthEndTime, dbAuthEndTime);
                    }
                }
            }
            log.info("分类结果，待新增/更新门店数: {}, 待删除商圈数: {}, 待更新授权时间的主门店数: {}",shopChangeTypeMap.get(ShopChangeTypeEnum.ADD).size(), deleteEsIds.size(), authEndTimeMap.size());

            // 三、数据转换，同步ES
            List<ShopInfo> insertShopInfoList = new ArrayList<>();

            // 使用策略模式，分别对四种类型的List<ShopVO> 转换成 List<ShopInfo>
            for (ShopChangeTypeEnum shopChangeTypeEnum : shopChangeTypeMap.keySet()) {
                List<ShopVO> shopVOList = shopChangeTypeMap.get(shopChangeTypeEnum);
                log.info("开始处理{}类型门店信息: {}", shopChangeTypeEnum.name(), JSON.toJSONString(shopVOList));

                // 根据shopChangeTypeEnum获取对应的转换策略
                ShopInfoConverterStrategy strategy = ShopInfoConverterStrategyFactory.getStrategy(shopChangeTypeEnum);
                try {
                    // 转换后的ES文档实体列表
                    List<ShopInfo> shopInfoList = strategy.convert(shopVOList, esShopInfoMap, merchantMap, saasShopConfigMap);
                    log.info("{}类型门店信息转换完成: {}", shopChangeTypeEnum.name(), JSON.toJSONString(shopInfoList));

                    if(ShopChangeTypeEnum.ADD.equals(shopChangeTypeEnum) && CollectionUtils.isNotEmpty(shopInfoList)){
                        // 新增的数据列表
                        insertShopInfoList.addAll(shopInfoList);
                    }
                } catch (Exception e) {
                    log.error("ES门店数据转换异常: " + shopChangeTypeEnum.name(), e);
                }
            }

            // 四、同步ES
            // 新增文档
            if (!insertShopInfoList.isEmpty()) {
                try {
                    shopInfoEsMapper.insertBatch(insertShopInfoList);
                } catch (Exception e) {
                    log.error("批量新增/更新ES门店信息异常:", e);
                }
            }

            // 删除文档
            if (!deleteEsIds.isEmpty()) {
                try {
                    shopInfoEsMapper.deleteBatchIds(deleteEsIds);
                } catch (Exception e) {
                    log.error("批量删除商圈信息异常", e);
                }
            }

            // 通过主门店id，批量更新子门店的授权结束时间
            if (!authEndTimeMap.isEmpty()) {
                try {
                    updateChidlShopAuthEndTime(authEndTimeMap);
                } catch (Exception e) {
                    log.error("批量更新子门店授权结束时间异常", e);
                }
            }

            long costTime = System.currentTimeMillis() - startTime;
            log.info("门店信息同步到ES完成，总耗时: {}ms，处理门店数: {}", costTime, dbShopInfos.size());
        } catch (BaseException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            // 释放锁
            if (lockAcquired) {
                redissonClient.getLock(RedisLockKeyConst.ES_SHOP_SYNC_LOCK).unlock();
                log.info("门店信息同步到ES，释放分布式锁");
            }
        }
    }

    /**
     * 通过主门店id，批量更新子门店的授权结束时间
     *
     * @param authEndTimeMap k=主门店id, v=授权到期时间
     */
    private void updateChidlShopAuthEndTime(Map<Long, Integer> authEndTimeMap) {
        for (Map.Entry<Long, Integer> entry : authEndTimeMap.entrySet()) {
            Long mainShopId = entry.getKey();
            Integer authEndTime = entry.getValue();

            try {
                // 获取索引名称
                String indexName = ShopInfo.class.getAnnotation(IndexName.class).value();
                // 创建UpdateByQueryRequest对象
                UpdateByQueryRequest request = new UpdateByQueryRequest(indexName);

                // 设置查询条件：parentId = mainShopId
                request.setQuery(QueryBuilders.termQuery("parent_id", mainShopId.toString()));

                // 设置更新脚本：设置auth_end_time字段
                Map<String, Object> params = new HashMap<>();
                params.put("auth_end_time", authEndTime);
                Script script = new Script(
                        ScriptType.INLINE,
                        "painless",
                        "ctx._source.auth_end_time = params.auth_end_time",
                        params);
                request.setScript(script);

                request.setAbortOnVersionConflict(false);

                // 日志记录更新前信息
                log.info("开始批量更新子门店授权结束时间，主门店ID: {}，授权结束时间: {}", mainShopId, authEndTime);

                // 同步至ES
                BulkByScrollResponse response = restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);

                // 详细记录更新结果
                if (response.getVersionConflicts() > 0) {
                    log.warn("批量更新子门店授权时间出现版本冲突，主门店ID: {}，冲突数量: {}", mainShopId, response.getVersionConflicts());
                }

                long totalDocs = response.getTotal();
                long updatedDocs = response.getUpdated();
                long timeTaken = response.getTook().getMillis();

                log.info("批量更新子门店授权结束时间完成，主门店ID: {}，授权结束时间: {}，总文档数: {}，成功更新数: {}，耗时: {}ms", mainShopId, authEndTime, totalDocs, updatedDocs, timeTaken);

                if (totalDocs > 0 && totalDocs != updatedDocs) {
                    log.warn("批量更新子门店授权结束时间，存在未成功更新的文档，主门店ID: {}，总数: {}，成功数: {}", mainShopId, totalDocs, updatedDocs);
                }

            } catch (Exception e) {
                log.error("批量更新子门店授权结束时间异常，主门店ID: " + mainShopId, e);
            }
        }
    }

    /**
     * 分批查询门店信息，避免大量门店ID查询时超时
     *
     * @param shopIds 门店ID列表
     * @return 所有门店信息的合并列表
     */
    private List<ShopVO> batchQueryShopInfoByIds(List<Long> shopIds) {
        if (CollectionUtils.isEmpty(shopIds)) {
            return Collections.emptyList();
        }

        // 结果列表
        List<ShopVO> resultList = new ArrayList<>();

        // 每批查询的大小
        final int batchSize = 10;

        List<List<Long>> subIdList = Lists.partition(shopIds, batchSize);
        int totalBatches = subIdList.size();
        for (int i = 0; i < totalBatches; i++) {
            List<Long> batchShopIds = subIdList.get(i);
            log.info("批次 {}/{} 查询门店信息，门店ID数量: {}", (i + 1), totalBatches, batchShopIds.size());
            try {
                // 调用接口查询当前批次
                Response<List<ShopVO>> batchResponse = shopFeign.listInfoByIds(batchShopIds);
                List<ShopVO> batchResult = ResponseUtil.getSucRes(batchResponse);

                if (CollectionUtils.isNotEmpty(batchResult)) {
                    resultList.addAll(batchResult);
                    log.info("批次 {}/{} 成功获取门店信息，数量: {}", (i + 1), totalBatches, batchResult.size());
                } else {
                    log.warn("批次 {}/{} 未查询到门店信息", (i + 1), totalBatches);
                }
            } catch (Exception e) {
                log.error("批次 {}/{} 查询门店信息异常: {}", (i + 1), totalBatches, e.getMessage(), e);
            }
        }

        return resultList;
    }
}
